package com.chinamobile.websocket.websocketdispatcher.distributor.redis;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.chinamobile.websocket.domain.dto.MessageDto;
import com.chinamobile.websocket.websocketdispatcher.common.MessageDistributorConstants;
import com.chinamobile.websocket.websocketdispatcher.distributor.AbstractMessageDistributor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;

/**
 * @className: RedisMessageDistributor
 * @description 基于 redis 的消息分发器, 订阅 websocket 发送消息，接收到消息时进行推送
 * @author: fanzhijun
 * @date: 2024/1/22
 **/

@Slf4j
public class RedisMessageDistributor<T extends MessageDto> extends AbstractMessageDistributor<T> implements MessageListener {
    private final StringRedisTemplate stringRedisTemplate;

    public RedisMessageDistributor(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    /**
     * @title 消息生产，由使用者调用
     * @author fanzhijun
     * @updateTime 2024/1/22 9:50
     */
    @Override
    public void distribute(T t) {
        MessageDto messageDto = new MessageDto();
        BeanUtils.copyProperties(t, messageDto);
        stringRedisTemplate.convertAndSend(MessageDistributorConstants.REDIS_WEBSOCKET_MESSAGE_CHANNEL, JSON.toJSONString(messageDto));
    }

    /**
     * @title 消息消费，redis订阅监听
     * @author fanzhijun
     * @updateTime 2024/1/22 9:51
     */
    @Override
    public void onMessage(Message message, byte[] bytes) {
        byte[] channelBytes = message.getChannel();
        RedisSerializer<String> stringSerializer = stringRedisTemplate.getStringSerializer();
        String channel = stringSerializer.deserialize(channelBytes);
        // 这里没有使用通配符，所以一定是true
        if (MessageDistributorConstants.REDIS_WEBSOCKET_MESSAGE_CHANNEL.equals(channel)) {
            byte[] bodyBytes = message.getBody();
            String body = stringSerializer.deserialize(bodyBytes);
            MessageDto messageDto = JSONObject.toJavaObject(JSON.parseObject(body), MessageDto.class);
            log.debug("send redis msg:" + JSON.toJSONString(messageDto));
            doSend(messageDto);
        }
    }

}
